package storm.online.bolt;

import java.util.Map;

import storm.online.common.LoginInfo;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class LogInfoParseBolt extends BaseRichBolt{

	private OutputCollector collector;
	
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		this.collector = collector;
	}

	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		String jsonStr = tuple.getStringByField("login_info");
//		System.out.println("debug string " + jsonStr);
		
		LoginInfo loginInfo = new LoginInfo();
		loginInfo.parseJson(jsonStr);
		
		System.out.println( "pt_id is " + loginInfo.getPtId() + ", game_id is " + loginInfo.getGameId() + 
							", area_id is " + loginInfo.getAreaId() + ", action is " + loginInfo.getAction() + 
							", timestamp is " + loginInfo.getTimeStamp() );
		
		collector.emit(new Values(loginInfo.getGameId(), loginInfo.getAreaId(), loginInfo.getPtId(), loginInfo.getAction(), loginInfo.getTimeStamp() ));
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		declarer.declare( new Fields("game_id", "area_id", "pt_id", "action", "timestamp"));
	}

}
